package com.atguigu.hbase;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnValueFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;

import java.io.IOException;

/**
 * @author yhm
 * @create 2023-02-09 18:57
 */
public class HBaseDML {
    public static Connection connection = HBaseConnect.getConnection();

    public static void putCell(String namespace,String tableName,String rowKey,String family,String column,String value) throws IOException {

        // 1. 获取table对象
        Table table = connection.getTable(TableName.valueOf(namespace,tableName));

        // 2. 调用对应的方法
        // 2.1 填写rowKey创建对象
        Put put = new Put(Bytes.toBytes(rowKey));

        // 2.2 添加cell值
        put.addColumn(Bytes.toBytes(family),Bytes.toBytes(column),Bytes.toBytes(value));
        try {
            table.put(put);
        } catch (Exception e) {
            System.out.println("写入数据有问题");
            e.printStackTrace();
        }

        // 3. 关闭table
        table.close();
    }

    public static void getCell(String namespace,String tableName,String rowKey,String family,String column) throws IOException {
        // 1. 获取table对象
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        // 2. 调用方法读取数据
        // 2.1 创建get对象填写rowKey
        Get get = new Get(Bytes.toBytes(rowKey));

        // 如果不写列名  默认读取一行数据
//        get.addColumn(Bytes.toBytes(family),Bytes.toBytes(column));

        // 多行cell
        Result result = null;
        try {
            result = table.get(get);
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                System.out.println(new String(CellUtil.cloneRow(cell)) + "-" + new String(CellUtil.cloneFamily(cell)) + "-" + new String(CellUtil.cloneQualifier(cell)) + "-" + new String(CellUtil.cloneValue(cell)));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }


        // 3. 关闭table
        table.close();

    }

    public static void scanRows(String namespace,String tableName,String startRow,String stopRow) throws IOException {
        // 1. 获取table对象
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        // 2. 调用方法读取数据
        // 2.1 创建scan对象
        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(startRow));
        // 添加第二个参数true可以控制右闭
        scan.withStopRow(Bytes.toBytes(stopRow));

        // 结果返回一个result的数组
        ResultScanner scanner = null;
        try {
            scanner = table.getScanner(scan);
            for (Result result : scanner) {
                for (Cell cell : result.rawCells()) {
                    System.out.print(new String(CellUtil.cloneRow(cell)) + "-" + new String(CellUtil.cloneFamily(cell)) + "-" + new String(CellUtil.cloneQualifier(cell)) + "-" + new String(CellUtil.cloneValue(cell)) + "\t");
                }
                System.out.println();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 3. 关闭table
        table.close();

    }

    public static void scanRowsWithFilter(String namespace,String tableName,String startRow,String stopRow,String family,String column,String value) throws IOException {
        // 1. 获取table对象
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        // 2. 调用方法读取数据
        // 2.1 创建scan对象
        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(startRow));
        // 添加第二个参数true可以控制右闭
        scan.withStopRow(Bytes.toBytes(stopRow));
        // columnValueFilter 最终只保留过滤的当前列数据
        ColumnValueFilter columnValueFilter = new ColumnValueFilter(
                Bytes.toBytes(family),
                Bytes.toBytes(column),
                CompareOperator.EQUAL,
                Bytes.toBytes(value)
        );

        // 保留整行的过滤器
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(family),
                Bytes.toBytes(column),
                CompareOperator.EQUAL,
                Bytes.toBytes(value)
        );
        // 可以使用多个过滤器
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        filterList.addFilter(singleColumnValueFilter);

        scan.setFilter(filterList);

        // 结果返回一个result的数组
        ResultScanner scanner = null;
        try {
            scanner = table.getScanner(scan);
            for (Result result : scanner) {
                for (Cell cell : result.rawCells()) {
                    System.out.print(new String(CellUtil.cloneRow(cell)) + "-" + new String(CellUtil.cloneFamily(cell)) + "-" + new String(CellUtil.cloneQualifier(cell)) + "-" + new String(CellUtil.cloneValue(cell)) + "\t");
                }
                System.out.println();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 3. 关闭table
        table.close();

    }

    public static void deleteCell(String namespace,String tableName,String rowKey,String family,String column) throws IOException {
        // 1. 获取table对象
        Table table = connection.getTable(TableName.valueOf(Bytes.toBytes(namespace), Bytes.toBytes(tableName)));

        // 2. 调用方法删除数据
        // 2.1 创建删除对象
        Delete delete = new Delete(Bytes.toBytes(rowKey));

        // addColumn 删除一个版本
        // addColumns 删除所有版本
        delete.addColumns(Bytes.toBytes(family), Bytes.toBytes(column));
        try {
            table.delete(delete);
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 3. 关闭table
        table.close();
    }

    public static void main(String[] args) {
        try {
//            putCell("bigdata","student","8888","info","age","18");
//            getCell("bigdata","student","8888","info","age");
//            scanRowsWithFilter("bigdata","student","1001","9999","info","name","galiman");
            deleteCell("bigdata","student","1001","info","age");
        } catch (IOException e) {

            System.out.println("连接获取关闭异常");
            e.printStackTrace();
        }

        System.out.println("其他程序");

        // 关闭连接
        HBaseConnect.closeConnect();
    }
}
